#include "rpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"
#include "logger.h"
#include "zookeeperutil.h"

#include <google/protobuf/descriptor.h>
#include <functional>

// 发布rpc方法的接口
void RpcProvider::NotifyService(::google::protobuf::Service *service)
{
    ServiceInfo service_info;

    // 服务对象的描述信息
    const google::protobuf::ServiceDescriptor *pserviceDesc = service->GetDescriptor();

    // 获取服务的名字(在.proto文件中service 类型定义的名字)
    std::string service_name = pserviceDesc->name();
    // std::cout << "service_name: " << service_name << std::endl;
    LOG_INFO("service_name: %s", service_name.c_str());

    // 获取服务对象的方法数量
    int methodCnt = pserviceDesc->method_count();

    for (int i = 0; i < methodCnt; i++)
    {
        // 获取服务对象指定下标的服务方法描述(抽象描述，也是一个指针)
        const google::protobuf::MethodDescriptor *pmethodDesc = pserviceDesc->method(i);
        std::string method_name = pmethodDesc->name();
        // std::cout << "method_name: " << method_name << std::endl;
        LOG_INFO("method_name: %s", method_name.c_str());
        service_info.m_methodMap.insert({method_name, pmethodDesc});
    }

    service_info.m_service = service;
    m_serviceMap.insert({service_name, service_info});
}

// 启动rpc服务节点，开始提供rpc远程网络调用服务
void RpcProvider::Run()
{
    std::string ip = MprpcApplication::GetInstance().GetConfig().GetValue("rpcserverip");
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().GetValue("rpcserverport").c_str());

    muduo::net::InetAddress address(ip, port);

    // 创建服务器对象
    muduo::net::TcpServer server(&m_eventloop, address, "RpcServer");

    // 设置相关回调函数包括消息读写回调
    server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this,
                                        std::placeholders::_1,
                                        std::placeholders::_2,
                                        std::placeholders::_3));

    // 设置线程数量
    server.setThreadNum(2);

    // 把当前rpc节点上的服务都注册到zk上面，让rpc客户端可以从zk上发现服务
    zkClient zkCli;
    zkCli.Start();
    // service_name为永久性节点，method_name为临时性节点
    for(auto &cp : m_serviceMap)
    {
        /*
            /server_name    e.g: /UserServiceRpc
        */
       std::string service_path = "/" + cp.first;
       zkCli.Create(service_path.c_str(), nullptr, 0);      // 第四个参数不传默认就是永久性节点

       for(auto &mp : cp.second.m_methodMap)
       {
            /*
                /service_name/method_name    e.g: /UserServiceRpc/Login
            */
           std::string method_path = service_path + "/" + mp.first;
           char data[128] = {0};
           sprintf(data, "%s:%d", ip.c_str(), port);

           // ZOO_EPHEMERAL表示znode为一个临时性节点
           zkCli.Create(method_path.c_str(), data, strlen(data), ZOO_EPHEMERAL);
       }
    }

    // std::cout << "RpcServer start at ip: " << ip << " , port: " << port << std::endl;
    LOG_INFO("RpcServer start at ip: %s, port: %d", ip.c_str(), port);

    // 启动网络服务
    server.start();
    m_eventloop.loop();
}

void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr &conn)
{
    // client 断开连接
    if (!conn->connected())
    {
        conn->shutdown();
    }
}

/*
    框架内部，RpcProvider和RpcConsumer协商好通信用的protobuf数据类型
    service_name method_name args   定义proto的message数据类型，进行数据头的序列化和反序列化
    e.g 19UserServiceRpcLoginzhang san123456
    header_size(4个字节) + header_str + args_str
*/
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
{
    std::string recv_buf = buf->retrieveAllAsString();

    // 读取字节流前4个字节
    uint32_t header_size;
    recv_buf.copy((char *)&header_size, 4, 0);

    // 根据header_size读取字节流的message类型
    std::string rpc_header_str = recv_buf.substr(4, header_size);

    // 反序列化
    mprpc::RpcHeader rpcheader;
    std::string service_name, method_name;
    uint32_t args_size;
    if (rpcheader.ParseFromString(rpc_header_str))
    {
        // 反序列化成功
        service_name = rpcheader.service_name();
        method_name = rpcheader.method_name();
        args_size = rpcheader.args_size();
    }
    else
    {
        // 反序列化失败
        // std::cout << "rpc_header_str: " << rpc_header_str << " parse error!" << std::endl;
        LOG_ERR("rpc_header_str: %s parse error!", rpc_header_str.c_str());
        return;
    }

    // 获取rpc方法参数的字节流数据
    std::string args_str = recv_buf.substr(4 + header_size, args_size);

    // 打印调试信息
    std::cout << "=========================================" << std::endl;
    std::cout << "header_size: " << header_size << std::endl;
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    std::cout << "service_name: " << service_name << std::endl;
    std::cout << "method_name: " << method_name << std::endl;
    std::cout << "args_size: " << args_size << std::endl;
    std::cout << "args_str: " << args_str << std::endl;
    std::cout << "=========================================" << std::endl;

    // 获取service对象和method对象
    auto it = m_serviceMap.find(service_name);
    if (it == m_serviceMap.end())
    {
        // std::cout << service_name << " is not exist!" << std::endl;
        LOG_ERR("%s is not exist!", service_name.c_str());
        return;
    }

    // 获取method对象
    auto mit = it->second.m_methodMap.find(method_name);
    if (mit == it->second.m_methodMap.end())
    {
        // std::cout << method_name << " is not exist!" << std::endl;
        LOG_ERR("%s is not exist!", method_name.c_str());
        return;
    }

    google::protobuf::Service *service = it->second.m_service;          // service对象
    const google::protobuf::MethodDescriptor *method = mit->second;     // method对象

    // 生成rpc方法调用的请求和响应参数
    google::protobuf::Message *request = service->GetRequestPrototype(method).New();
    if(!request->ParseFromString(args_str))
    {
        // std::cout << "request parse error, content: " << args_str << std::endl;
        LOG_ERR("request parse error, content: %s", args_str.c_str());
        return;
    }

    google::protobuf::Message *response = service->GetResponsePrototype(method).New();

    // 给下面的调用绑定一个Closure回调方法
    google::protobuf::Closure *done = google::protobuf::NewCallback<RpcProvider, const muduo::net::TcpConnectionPtr&, google::protobuf::Message*>(this, &RpcProvider::SendRpcResponse, conn, response);

    // 调用当前的rpc方法
    // 相当于new UserService().Login(controller, request, response, done)
    service->CallMethod(method, nullptr, request, response, done);
}

void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    std::string resp_str;
    if(response->SerializeToString(&resp_str))
    {
        // 序列化成功后，通过网络把resp发送给rpc调用方
        conn->send(resp_str);
    }
    else
    {
        // std::cout << "serialize response error! " << std::endl;
        LOG_ERR("serialize response error! ");
    }
    conn->shutdown();   // 模拟HTTP短链接，请求一次即关闭连接，由RpcProvider主动断开连接
}
